package com.raft.core.node.task;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import javax.annotation.Nonnull;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.raft.core.log.statemachine.StateMachine;
import com.raft.core.node.NodeEndpoint;
import com.raft.core.node.NodeId;

public class ReadIndexTask {

	private static final Logger logger = LoggerFactory.getLogger(ReadIndexTask.class);
	private final int commitIndex;
	private final Map<NodeId, Integer> matchIndices;
	private final StateMachine stateMachine;
	private final String requestId;

	public ReadIndexTask(int commitIndex, Set<NodeEndpoint> followerIds, StateMachine stateMachine, String requestId) {
		this.commitIndex = commitIndex;
		this.stateMachine = stateMachine;
		this.requestId = requestId;

		matchIndices = new HashMap<>();
		for (NodeEndpoint nodeEndpoint : followerIds) {
			NodeId nodeId = nodeEndpoint.getId();
			matchIndices.put(nodeId, 0);
		}
	}

	/**获取请求id
	 * @return
	 */
	public String getRequestId() {
		return requestId;
	}

	/**更新节点的matchIndex，然后判断是否过半，返回值表示是否过半
	 * @param nodeId
	 * @param matchIndex
	 * @return
	 */
	public boolean updateMatchIndex(@Nonnull NodeId nodeId, int matchIndex) {
		logger.info("update match index, node id {}, match index {}", nodeId, matchIndex);
		if(!matchIndices.containsKey(nodeId)) {
			// 集群成员可能会有变化，ReadIndex无法阻止集群成员变更，而且集群成员变更下的ReadIndex存在不确定性。
			// 一个简单的策略是，遇到不在期望列表中的节点时抛出异常，然后告知客户端集群成员发送的变化。另一个策略是，集群成员变更时通知所有ReadIndx任务，
			// 让任务失败。本书为了简单，碰到这种情况，直接返回false
			return false;
		}
		matchIndices.put(nodeId, matchIndex);
		long countOfFollowerReachingCommitIndex = matchIndices.values().stream().filter(i -> i >= commitIndex).count();
		// 是否过半
		if(countOfFollowerReachingCommitIndex + 1 > (matchIndices.size() + 1)/2) {
			stateMachine.onReadIndexReached(requestId, commitIndex);
			return true;
		}
		return false;
	}

	@Override
	public String toString() {
		return "ReadIndexTask [commitIndex=" + commitIndex + ", matchIndices=" + matchIndices + ", requestId=" + requestId + "]";
	}

}
